package com.bw.gmall.realtime.common.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.HBaseUtil;
import com.bw.gmall.realtime.common.util.RedisUtil;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public abstract  class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements IDimAsync<T>  {

    String table;
    private AsyncConnection hBaseAsyncConnection;
    private StatefulRedisConnection<String, String> redisAsyncConnection;

    public DimAsyncFunction(String table) {
        this.table=table;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //获取 hbase 异步
        hBaseAsyncConnection = HBaseUtil.getHBaseAsyncConnection();

        //获取redis异步
        redisAsyncConnection = RedisUtil.getRedisAsyncConnection();
    }

    @Override
    public void close() throws Exception {

        HBaseUtil.closeAsyncConnection(hBaseAsyncConnection);

        RedisUtil.closeRedisAsyncConnection(redisAsyncConnection);

    }

    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture) throws Exception {

        String dimId = getDimId(t);


        String redisKey = RedisUtil.getRedisKey(table, dimId);

        //先进行 查询 redis 是否有数据

        CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override
            public String get() {
                String s =null;
                RedisFuture<String> redisFuture = redisAsyncConnection.async().get(redisKey);

                try {

                    s=redisFuture.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }

                return s;
            }
        }).thenApplyAsync(new Function<String, JSONObject>() {
            @Override
            public JSONObject apply(String s) {
                JSONObject jsonObject =null;
                //先查询redis
                if(s==null || s.length()==0){

                    //从hbase取值
                    try {
                        jsonObject = HBaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, table, dimId);

                    //存到 redis 中
                        redisAsyncConnection.async().setex(redisKey,24*3600,jsonObject.toJSONString());
                    } catch (IOException e) {
                        e.printStackTrace();
                    }

                }else{

                    //从 redis  中取值
                    jsonObject= JSON.parseObject(s);
                }

                return jsonObject;
            }
        }).thenAccept(new Consumer<JSONObject>() {
            @Override
            public void accept(JSONObject jsonObject) {
                //取值 存值

                if(jsonObject!=null){

                    setDim(t,jsonObject);
                }else{
                    System.out.println("没有查到维度数据:" + table + ":" + dimId);
                }

                //返回结果

                resultFuture.complete(Collections.singletonList(t));
            }
        });
    }

}
